package com.raptor.logs.delivery;

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.TimeoutException;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @since 0.0.1
 */
public class AsynchronousDeliveryStrategy implements DeliveryStrategy {

    ExecutorService executorService = Executors.newFixedThreadPool(2);

    @Override
    public <K, V, E> Future<Boolean> send(Producer<K, V> producer, ProducerRecord<K, V> record, final E event,
                 final FailedDeliveryCallback<E> failedDeliveryCallback) {
        Future<Boolean> f = executorService.submit(()->{
            try {
                producer.send(record, (metadata, exception) -> {
                    if (exception != null) {
                        failedDeliveryCallback.onFailedDelivery(event, exception);
                    }
                });
                return true;
            } catch (TimeoutException e) {
                failedDeliveryCallback.onFailedDelivery(event, e);
                return false;
            }
        });

        return f;
    }

}
